package org.lds.simple.disruptor.consumer;

import org.lds.simple.disruptor.*;
import org.lds.simple.disruptor.consumer.exception.ConsumeExceptionHandler;
import org.lds.simple.disruptor.consumer.exception.IgnoreConsumeExceptionHandler;
import org.lds.simple.disruptor.producer.sequencer.Sequencer;
import org.lds.simple.disruptor.sequence.Sequence;
import org.lds.simple.disruptor.strategy.WaitStrategy;

import java.util.ArrayList;
import java.util.List;

public class EventConsumerProcessor implements LifeCycle {
    private final Sequencer sequencer;
    private final RingBuffer ringBuffer;
    private final WaitStrategy waitStrategy;

    private static final ConsumeExceptionHandler DEFAULT_CONSUME_EXCEPTION_HANDLER = new IgnoreConsumeExceptionHandler();
    private final ConsumeExceptionHandler consumeExceptionHandler;

    private List<EventConsumer> eventConsumers = new ArrayList<>();

    public EventConsumerProcessor(Sequencer sequencer, RingBuffer ringBuffer, WaitStrategy waitStrategy) {
        this(sequencer, ringBuffer, waitStrategy, null);
    }

    public EventConsumerProcessor(Sequencer sequencer, RingBuffer ringBuffer, WaitStrategy waitStrategy, ConsumeExceptionHandler consumeExceptionHandler) {
        this.sequencer = sequencer;
        this.ringBuffer = ringBuffer;
        this.waitStrategy = waitStrategy;
        this.consumeExceptionHandler = buildConsumeExceptionHandler(consumeExceptionHandler);
    }

    private ConsumeExceptionHandler buildConsumeExceptionHandler(ConsumeExceptionHandler consumeExceptionHandler) {
        if (consumeExceptionHandler == null) return DEFAULT_CONSUME_EXCEPTION_HANDLER;
        return consumeExceptionHandler;
    }


    public void register(Class<? extends EventConsumer> consumerClass) {
        try {
            EventConsumer eventConsumer = consumerClass.newInstance();
            eventConsumers.add(eventConsumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void start() {
        EventConsumerTask[] eventConsumerTasks = getEventConsumerTasks();

        Sequence[] consumerSequences = getConsumerSequences(eventConsumerTasks);

        SequenceBarrier sequenceBarrier = new SequenceBarrier(ringBuffer.getBufferSize(), sequencer.getSequence(), consumerSequences, waitStrategy);
        sequencer.setSequenceBarrier(sequenceBarrier);

        for (EventConsumerTask consumerTask : eventConsumerTasks) {
            new Thread(consumerTask).start();
        }
    }

    private Sequence[] getConsumerSequences(EventConsumerTask[] eventConsumerTasks) {
        Sequence[] consumerSequences = new Sequence[eventConsumerTasks.length];
        for (int i = 0; i < eventConsumerTasks.length; i++) {
            consumerSequences[i] = eventConsumerTasks[i].getSequence();
        }
        return consumerSequences;
    }

    private EventConsumerTask[] getEventConsumerTasks() {
        EventConsumerTask[] eventConsumerTasks = new EventConsumerTask[eventConsumers.size()];
        for (int i = 0; i < eventConsumers.size(); i++) {
            eventConsumerTasks[i] = new EventConsumerTask(sequencer, ringBuffer, eventConsumers.get(i), waitStrategy, consumeExceptionHandler);
        }
        return eventConsumerTasks;
    }

    @Override
    public void shutdown() {
        EventConsumerTask[] eventConsumerTasks = getEventConsumerTasks();
        for (int i = 0; i < eventConsumerTasks.length; i++) {
            eventConsumerTasks[i].shutdown();
        }
    }
}
